package com.xrui.hbase.filter;

import com.google.common.base.Objects;
import com.xrui.hbase.DataRequest;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;
import java.util.regex.Pattern;

/**
 * A RowFilter that only allows qualifiers that match a given regular expression.
 */
public final class RegexRowFilter extends RowFilter {
    private static final long serialVersionUID = 1L;

    private final String mRegularExpression;

    /**
     * Constructor.
     *
     * @param regularExpression The regular expression for qualifiers that should be
     *                          accepted by this filter. The expression is matched against the full qualifier (as
     *                          if it implicitly starts with '^' and ends with '$'.
     */
    public RegexRowFilter(String regularExpression) {
        // Try to compile the regular expression, so clients create an early error message.
        Pattern.compile(regularExpression);  // <-- throws a PatternSyntaxException if invalid.

        mRegularExpression = regularExpression;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter() throws IOException {
        return new org.apache.hadoop.hbase.filter.RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(mRegularExpression));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof RegexRowFilter)) {
            return false;
        } else {
            final RegexRowFilter otherFilter = (RegexRowFilter) other;
            return Objects.equal(otherFilter.mRegularExpression, mRegularExpression);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return mRegularExpression.hashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public DataRequest getDataRequest() {
        return DataRequest.builder().build();
    }


    /**
     * {@inheritDoc}
     */
    @Override
    protected JsonNode toJsonNode() {
        final ObjectNode root = JsonNodeFactory.instance.objectNode();
        root.put("regex", mRegularExpression);
        return root;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected Class<? extends RegexRowFilterDeserializer> getDeserializerClass() {
        return RegexRowFilter.RegexRowFilterDeserializer.class;
    }

    /**
     * Deserializes {@code HasColumnDataRowFilter}.
     */
    public static final class RegexRowFilterDeserializer
        implements RowFilterDeserializer {
        /**
         * {@inheritDoc}
         */
        @Override
        public RowFilter createFromJson(JsonNode root) {
            final String regex = root.path("regex").getTextValue();
            return new RegexRowFilter(regex);
        }
    }
}
